/*
 * TcpMediaCasterConnection.cpp
 *
 *  Created on: 2019年3月1日
 *      Author: terry
 */

#include "TcpMediaCasterConnection.h"

#include "CLog.h"
#include "TStringUtil.h"
#include "Socket.h"
#include <assert.h>
#include <StopWatch.h>
#include "MediaCasterConfig.h"


namespace av
{


TcpMediaCasterConnection::TcpMediaCasterConnection():
    m_streamState(av::STATE_STOPPED),
    m_jamming(),
    m_maxCacheSize(),
    m_empty()
{
    m_maxCacheSize = 1024 * 1000;
}

TcpMediaCasterConnection::~TcpMediaCasterConnection()
{
}


int TcpMediaCasterConnection::onOpen()
{
    bufferevent* bev = getBufferEvent();
    bufferevent_setwatermark(bev, EV_WRITE, 1024 * 16, m_maxCacheSize);

    //bufferevent_enable(bev, EV_WRITE);

    evutil_socket_t fd = bufferevent_getfd(bev);
    evutil_make_socket_nonblocking(fd);

    //int value = true;
    //int ret = ::setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char*)&value, sizeof(value));

    comn::Socket sock;
    sock.attach(fd);
    if (MediaCasterConfig::SEND_BUFFER_SIZE > 0)
    {
        sock.setSendBufferSize(MediaCasterConfig::SEND_BUFFER_SIZE);
    }

    comn::SockAddr peerAddr = sock.getPeerName();
    m_peerID = comn::StringUtil::format("%s:%d", peerAddr.getIPAddr(), peerAddr.getPort());
    sock.detach();

    return 0;
}

void TcpMediaCasterConnection::onClose()
{
    CLog::debug("TcpMediaCasterConnection(%s) onClose stream%s\n",
        m_peerID.c_str(),
        m_streamName.c_str());

}

void TcpMediaCasterConnection::onWrite(struct bufferevent *bev)
{
    //
}

void TcpMediaCasterConnection::onRead(struct bufferevent *bev)
{
    // pass
}

void TcpMediaCasterConnection::onEvent(struct bufferevent *bev, short events)
{
    TcpConnection::onEvent(bev, events);
}

void TcpMediaCasterConnection::setStreamName(const std::string& name)
{
    m_streamName = name;
}

void TcpMediaCasterConnection::onMediaFormat(const av::MediaFormat& fmt)
{
    m_format = fmt;
}

void TcpMediaCasterConnection::onMediaPacket(av::MediaPacketPtr& pkt)
{
    comn::StopWatch watch;
    writeMedia(pkt);

    uint64_t elapse = watch.elapse();
    if (elapse > 20)
    {
        char pktType = pkt->isVideo() ? 'V' : 'A';
        CLog::warning("TcpMediaCasterConnection::onMediaPacket elapse too much. pkt type:%c, elapse:%d\n", pktType, elapse);
    }
}

void TcpMediaCasterConnection::onMediaEvent(int event)
{
    // pass
}

int TcpMediaCasterConnection::writeMedia(av::MediaPacketPtr& pkt)
{
    bufferevent* bev = getBufferEvent();
    if (!bev)
    {
        return 0;
    }

    uint8_t* buf = pkt->data;
    int buf_size = pkt->size;

    evbuffer * evbuf = bufferevent_get_output(bev);
    size_t length = evbuffer_get_length(evbuf);

    if (length > m_maxCacheSize)
    {
        if (m_jamming)
        {
            /// 拥塞中, 不追加.
            size_t count = m_pktQueue.dropUntilKeyFrame();
            CLog::warning("evbuffer cache is too much. cache:%d, drop:%d\n", length, count);
        }
        else
        {
            m_jamming = true;
            CLog::warning("TcpMediaCasterConnection(%s) is jamming. buffer size: %d\n", m_peerID.c_str(), length);
            bufferevent_enable(bev, EV_WRITE);
        }
    }
    else if (length > m_maxCacheSize/4)
    {
        if (m_jamming)
        {
            /// 拥塞中, 不追加.
        }
        else
        {
            bufferevent_write(bev, (char*)buf, buf_size);
        }
    }
    else
    {
        if (m_jamming)
        {
            m_jamming = false;
            CLog::warning("TcpMediaCasterConnection(%s) resume from jam: buffer size%d\n", m_peerID.c_str(), length);
        }

        bufferevent_write(bev, (char*)buf, buf_size);
    }

    return buf_size;
}


} /* namespace av */
